package com.csap.framework.integration.listener;

import com.alibaba.fastjson.JSON;
import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.csap.framework.core.ApplicationContextHolder;
import com.csap.framework.integration.MessageAction;
import com.csap.framework.integration.RocketmqConfig;
import com.csap.framework.integration.consumer.EnableRocketmqBean;
import com.csap.framework.util.FinalConfigParam;
import com.csap.framework.util.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.SerializationUtils;

import java.lang.reflect.ParameterizedType;

/**
 * @author yangchengfu
 * @description 消息监听
 * @dataTime 2020年-10月-12日 17:14:00
 **/
public interface MessageListener<T> extends com.aliyun.openservices.ons.api.MessageListener {
    Logger log = LoggerFactory.getLogger(MessageListener.class);

    /**
     * 消费消息接口，由应用来实现<br>
     * 网络抖动等不稳定的情形可能会带来消息重复，对重复消息敏感的业务可对消息做幂等处理
     *
     * @param message 消息
     * @param context 消费上下文
     * @param t       返回的泛型具体数据
     */
    void consume(final Message message, final T t, final ConsumeContext context);

    /**
     * 消费消息接口，由应用来实现<br>
     * 网络抖动等不稳定的情形可能会带来消息重复，对重复消息敏感的业务可对消息做幂等处理
     *
     * @param message 消息
     * @param context 消费上下文
     * @return 消费结果，如果应用抛出异常或者返回Null等价于返回Action.ReconsumeLater
     * @see <a href="https://help.aliyun.com/document_detail/44397.html">如何做到消费幂等</a>
     */
    @Override
    default Action consume(Message message, ConsumeContext context) {
        log.debug("MessageListener:key {} {} {}", message.getKey(), message.getMsgID(), message.getTopic());
        RocketmqConfig rocketmqConfig = ApplicationContextHolder.getBean(RocketmqConfig.class);
        try {
            EnableRocketmqBean enableRocketmqBean = ApplicationContextHolder.getBean(EnableRocketmqBean.class);
            EnableRocketmqBean.RocketmqBean rocketmqBean = enableRocketmqBean.get(this.getClass().getName());
            if (StringUtil.isNotEmpty(message.getKey()) && StringUtil.isNotEmpty(rocketmqBean.getKey())) {
                if (!message.getKey().equals(rocketmqBean.getKey())) {
                    return Action.CommitMessage;
                }
            }
            switch (rocketmqConfig.getConsumer().getMessageType()) {
                case JSON:
                    consume(message, JSON.parseObject(message.getBody(), getMessageClass()), context);
                    break;
                case STRING:
                    consume(message, (T) new String(message.getBody(), FinalConfigParam.DEFAULT_CHARSET), context);
                    break;
                default:
                    consume(message, (T) SerializationUtils.deserialize(message.getBody()), context);
            }
            return Action.CommitMessage;
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            return rocketmqConfig.getConsumer().equals(MessageAction.ERROT) ? Action.ReconsumeLater : Action.CommitMessage;
        }
    }

    /**
     * 获取消息具体对象
     *
     * @return
     */
    default Class<T> getMessageClass() {
        return (Class<T>) ((ParameterizedType) this.getClass().getGenericInterfaces()[0]).getActualTypeArguments()[0];
    }
}
